Node / meshtastic / Meshtastic-Android / files / core / network / src / commonMain / kotlin / org / meshtastic / core / network / radio / ReplayRadioTransport.kt

Displaying Raw • Download

T8b949e/*

* Copyright (c) 2026 Meshtastic LLC

*

* This program is free software: you can redistribute it and/or modify

* it under the terms of the GNU General Public License as published by

* the Free Software Foundation, either version 3 of the License, or

* (at your option) any later version.

*

* This program is distributed in the hope that it will be useful,

* but WITHOUT ANY WARRANTY; without even the implied warranty of

* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

* GNU General Public License for more details.

*

* You should have received a copy of the GNU General Public License

* along with this program. If not, see <https://www.gnu.org/licenses/>.

*/

Tff7b72package T7ee787org.meshtastic.core.network.radio

Tff7b72import T7ee787co.touchlab.kermit.Logger

Tff7b72import T7ee787kotlinx.coroutines.CoroutineScope

Tff7b72import T7ee787kotlinx.coroutines.delay

Tff7b72import T7ee787okio.Buffer

Tff7b72import T7ee787okio.EOFException

Tff7b72import T7ee787org.meshtastic.core.common.util.handledLaunch

Tff7b72import T7ee787org.meshtastic.core.repository.HandshakeConstants

Tff7b72import T7ee787org.meshtastic.core.repository.RadioTransport

Tff7b72import T7ee787org.meshtastic.core.repository.RadioTransportCallback

Tff7b72import T7ee787org.meshtastic.proto.FromRadio

Tff7b72import T7ee787org.meshtastic.proto.ToRadio

T8b949e/**

* A [RadioTransport] that replays a pre-captured stream of `FromRadio` frames entirely on-device — no network and no

* paired radio. It is the deterministic, self-contained traffic source behind the "Demo Mode (Replay)" connection

* entry, used to drive realistic ~200-node mesh traffic into the app for Macrobenchmark / Baseline Profile journeys and

* populated-UI (node list / map / message) tests.

*

* ### What it exercises

* Frames are handed verbatim to [RadioTransportCallback.handleFromRadio] — the *same* ingestion entry a live

* BLE/TCP/Serial radio uses — so everything downstream (protobuf decode, the want_config state machine, node-DB and

* packet handling, the UI) runs exactly as in production. The only things it skips are the physical link and the stream

* framing layer: it deals in already-deframed `FromRadio` payloads.

*

* ### Asset format (`*.fromradio`)

* A flat, three-section container. **Every integer is a big-endian (network-order) unsigned 32-bit value:**

*

* ```

* ┌─ Stage-1: config section ────────────────────────────────────────────┐

* │ u32 C number of config frames │

* │ C × ( u32 len, len bytes ) each blob = one serialized FromRadio │

* ├─ Stage-2: node section ──────────────────────────────────────────────┤

* │ u32 N number of node_info frames │

* │ N × ( u32 len, len bytes ) each blob = one serialized FromRadio │

* ├─ packet section ─────────────────────────────────────────────────────┤

* │ ( u32 len, len bytes ) * repeated until end-of-buffer │

* └──────────────────────────────────────────────────────────────────────┘

* ```

*

* Each `len`-prefixed blob is a single `FromRadio` message as produced by Wire's `encode()`. By section:

* - **config** — `my_info`, `metadata`, `config.*`, `moduleConfig.*`, and `channel` frames: everything the app needs to

* come up, but **no** `node_info` (the app ignores NodeInfo during Stage 1).

* - **node** — the connected node first, then the captured node DB (one `node_info` per frame).

* - **packet** — the captured `MeshPacket` traffic, in capture order; the section has no count and runs to EOF.

*

* `config_complete_id` is deliberately **omitted** from the file — this transport injects it per phase (below) with the

* app's own nonce, exactly as real firmware echoes the request.

*

* The file is produced by the burningmesh-replay tool (`replay_server.py --export PATH`, which also sanitizes capture

* PII). **This class is the authoritative reader of the format; keep the exporter and [ReplayRadioTransportTest]'s

* `asset()` encoder in sync with the layout above.**

*

* ### Handshake & runtime behaviour

* Driven by the app's two-phase want_config handshake ([HandshakeConstants]):

* - **Stage 1** ([HandshakeConstants.CONFIG_NONCE]) → emit the config section, then `config_complete_id`.

* - **Stage 2** ([HandshakeConstants.NODE_INFO_NONCE]) → emit the node section, then `config_complete_id`, then start

* streaming the packet section **once** (a re-issued Stage-2 request does not restart it).

*

* The packet stream is paced at [packetDelayMs] and does not loop; it stops when [scope] is cancelled. Outbound

* [ToRadio] traffic other than want_config (heartbeats, app packets) is ignored — the replay is strictly read-only, and

* frames are held in memory, so [close] has nothing to release.

*

* ### Robustness

* The asset is parsed up front and every length is validated against the bytes actually remaining, so a truncated or

* corrupt file fails fast with [IllegalArgumentException] rather than a cryptic buffer underflow or an out-of-memory

* allocation. [handleSendToRadio] likewise tolerates any inbound bytes (undecodable [ToRadio] is ignored, not thrown).

* The bundled asset is trusted, but this same reader is a convenient injection point for malformed-input / fuzz testing

* of the ingestion path — see `ReplayFuzz` in the test sources.

*/

Tff7b72class T56d364ReplayRadioTransportTb4b4b4(

Tff7b72private Tff7b72val Te6edf3callbackTb4b4b4: Te6edf3RadioTransportCallbackTb4b4b4,

Tff7b72private Tff7b72val Te6edf3scopeTb4b4b4: Te6edf3CoroutineScopeTb4b4b4,

Tff7b72val Te6edf3addressTb4b4b4: Tffa657StringTb4b4b4,

Te6edf3framesTb4b4b4: Te6edf3ByteArrayTb4b4b4,

Tff7b72private Tff7b72val Te6edf3packetDelayMsTb4b4b4: Tffa657Long Tff7b72= Te6edf3DEFAULT_PACKET_DELAY_MSTb4b4b4,

Tb4b4b4) Tb4b4b4: Te6edf3RadioTransport Tb4b4b4{

Tff7b72private Tff7b72val Te6edf3configFramesTb4b4b4: Te6edf3ListTff7b72<Te6edf3ByteArrayTff7b72>

Tff7b72private Tff7b72val Te6edf3nodeFramesTb4b4b4: Te6edf3ListTff7b72<Te6edf3ByteArrayTff7b72>

Tff7b72private Tff7b72val Te6edf3packetFramesTb4b4b4: Te6edf3ListTff7b72<Te6edf3ByteArrayTff7b72>

Tff7b72init Tb4b4b4{

Tff7b72val Te6edf3buffer Tff7b72= Te6edf3BufferTb4b4b4(Tb4b4b4)Tb4b4b4.Te6edf3writeTb4b4b4(Te6edf3framesTb4b4b4)

Te6edf3configFrames Tff7b72= Te6edf3bufferTb4b4b4.Te6edf3readSectionTb4b4b4(Te6edf3counted Tff7b72= Tff7b72trueTb4b4b4, Te6edf3label Tff7b72= Ta5d6ff"Ta5d6ffconfigTa5d6ff"Tb4b4b4)

Te6edf3nodeFrames Tff7b72= Te6edf3bufferTb4b4b4.Te6edf3readSectionTb4b4b4(Te6edf3counted Tff7b72= Tff7b72trueTb4b4b4, Te6edf3label Tff7b72= Ta5d6ff"Ta5d6ffnodeTa5d6ff"Tb4b4b4)

Te6edf3packetFrames Tff7b72= Te6edf3bufferTb4b4b4.Te6edf3readSectionTb4b4b4(Te6edf3counted Tff7b72= Tff7b72falseTb4b4b4, Te6edf3label Tff7b72= Ta5d6ff"Ta5d6ffpacketTa5d6ff"Tb4b4b4)

Tb4b4b4}

Tff7b72private Tff7b72var Te6edf3packetsStarted Tff7b72= Tff7b72false

Tff7b72override Tff7b72fun Td2a8ffstartTb4b4b4(Tb4b4b4) Tb4b4b4{

Te6edf3LoggerTb4b4b4.Te6edf3i Tb4b4b4{

Ta5d6ff"Ta5d6ffStarting replay transport: Tffd700${Te6edf3configFramesTb4b4b4.Te6edf3sizeTffd700}Ta5d6ff config, Tffd700${Te6edf3nodeFramesTb4b4b4.Te6edf3sizeTffd700}Ta5d6ff node, Ta5d6ff" Tff7b72+

Ta5d6ff"Tffd700${Te6edf3packetFramesTb4b4b4.Te6edf3sizeTffd700}Ta5d6ff packet framesTa5d6ff"

Tb4b4b4}

Te6edf3callbackTb4b4b4.Te6edf3onConnectTb4b4b4(Tb4b4b4)

Tb4b4b4}

Tff7b72override Tff7b72fun Td2a8ffhandleSendToRadioTb4b4b4(Te6edf3pTb4b4b4: Te6edf3ByteArrayTb4b4b4) Tb4b4b4{

T8b949e// Undecodable ToRadio is ignored rather than thrown: the replay must tolerate any bytes the app — or a fuzz

T8b949e// harness — hands it, exactly as it tolerates a malformed asset.

Tff7b72val Te6edf3wantConfigId Tff7b72= Te6edf3runCatching Tb4b4b4{ Te6edf3ToRadioTb4b4b4.Te6edf3ADAPTERTb4b4b4.Te6edf3decodeTb4b4b4(Te6edf3pTb4b4b4)Tb4b4b4.Te6edf3want_config_id Tb4b4b4}Tb4b4b4.Te6edf3getOrNullTb4b4b4(Tb4b4b4)

Tff7b72when Tb4b4b4(Te6edf3wantConfigIdTb4b4b4) Tb4b4b4{

Te6edf3HandshakeConstantsTb4b4b4.Te6edf3CONFIG_NONCE Tff7b72-Tff7b72>

Te6edf3scopeTb4b4b4.Te6edf3handledLaunch Tb4b4b4{

Te6edf3emitTb4b4b4(Te6edf3configFramesTb4b4b4)

Te6edf3completeTb4b4b4(Te6edf3HandshakeConstantsTb4b4b4.Te6edf3CONFIG_NONCETb4b4b4)

Tb4b4b4}

Te6edf3HandshakeConstantsTb4b4b4.Te6edf3NODE_INFO_NONCE Tff7b72-Tff7b72>

Te6edf3scopeTb4b4b4.Te6edf3handledLaunch Tb4b4b4{

Te6edf3emitTb4b4b4(Te6edf3nodeFramesTb4b4b4)

Te6edf3completeTb4b4b4(Te6edf3HandshakeConstantsTb4b4b4.Te6edf3NODE_INFO_NONCETb4b4b4)

Tff7b72if Tb4b4b4(Tff7b72!Te6edf3packetsStartedTb4b4b4) Tb4b4b4{

Te6edf3packetsStarted Tff7b72= Tff7b72true

Te6edf3streamPacketsTb4b4b4(Tb4b4b4)

Tb4b4b4}

Tb4b4b4}

T8b949e// All other ToRadio traffic (heartbeats, outbound packets) is ignored — this is a read-only replay.

Tb4b4b4}

Tb4b4b4}

Tff7b72private Tff7b72fun Td2a8ffemitTb4b4b4(Te6edf3framesTb4b4b4: Te6edf3ListTff7b72<Te6edf3ByteArrayTff7b72>Tb4b4b4) Tff7b72= Te6edf3framesTb4b4b4.Te6edf3forEach Tb4b4b4{ Te6edf3callbackTb4b4b4.Te6edf3handleFromRadioTb4b4b4(Tffa657itTb4b4b4) Tb4b4b4}

Tff7b72private Tff7b72fun Td2a8ffcompleteTb4b4b4(Te6edf3nonceTb4b4b4: Tffa657IntTb4b4b4) Tff7b72= Te6edf3callbackTb4b4b4.Te6edf3handleFromRadioTb4b4b4(Te6edf3FromRadioTb4b4b4(Te6edf3config_complete_id Tff7b72= Te6edf3nonceTb4b4b4)Tb4b4b4.Te6edf3encodeTb4b4b4(Tb4b4b4)Tb4b4b4)

Tff7b72private Tff7b72suspend Tff7b72fun Td2a8ffstreamPacketsTb4b4b4(Tb4b4b4) Tb4b4b4{

Te6edf3LoggerTb4b4b4.Te6edf3d Tb4b4b4{ Ta5d6ff"Ta5d6ffReplay streaming Tffd700${Te6edf3packetFramesTb4b4b4.Te6edf3sizeTffd700}Ta5d6ff packets at Tffd700${Te6edf3packetDelayMsTffd700}Ta5d6ffms spacingTa5d6ff" Tb4b4b4}

Tff7b72for Tb4b4b4(Te6edf3frame Tff7b72in Te6edf3packetFramesTb4b4b4) Tb4b4b4{

Te6edf3callbackTb4b4b4.Te6edf3handleFromRadioTb4b4b4(Te6edf3frameTb4b4b4)

Tff7b72if Tb4b4b4(Te6edf3packetDelayMs Tff7b72> T79c0ff0Tb4b4b4) Te6edf3delayTb4b4b4(Te6edf3packetDelayMsTb4b4b4)

Tb4b4b4}

Te6edf3LoggerTb4b4b4.Te6edf3i Tb4b4b4{ Ta5d6ff"Ta5d6ffReplay finished (Tffd700${Te6edf3packetFramesTb4b4b4.Te6edf3sizeTffd700}Ta5d6ff packets)Ta5d6ff" Tb4b4b4}

Tb4b4b4}

Tff7b72override Tff7b72suspend Tff7b72fun Td2a8ffcloseTb4b4b4(Tb4b4b4) Tb4b4b4{

T8b949e// Frames live in memory; the streaming coroutine is cancelled with the scope.

Tb4b4b4}

T8b949e/**

* Reads one section of the asset. A [counted] section is prefixed with a `u32` frame count; the final (packet)

* section is uncounted and runs to end-of-buffer. Every frame length is checked against the bytes still available,

* so a malformed/truncated asset throws [IllegalArgumentException] up front instead of underflowing the buffer or

* allocating a multi-gigabyte array from a bogus length.

*/

Tff7b72private Tff7b72fun Te6edf3BufferTb4b4b4.Td2a8ffreadSectionTb4b4b4(Te6edf3countedTb4b4b4: Tffa657BooleanTb4b4b4, Te6edf3labelTb4b4b4: Tffa657StringTb4b4b4)Tb4b4b4: Te6edf3ListTff7b72<Te6edf3ByteArrayTff7b72> Tb4b4b4{

Tff7b72val Te6edf3expected Tff7b72= Tff7b72if Tb4b4b4(Te6edf3countedTb4b4b4) Te6edf3readUInt32Tb4b4b4(Ta5d6ff"Tffd700$Te6edf3labelTa5d6ff frame countTa5d6ff"Tb4b4b4) Tff7b72else Tffa657IntTb4b4b4.Te6edf3MAX_VALUE

Tff7b72val Te6edf3frames Tff7b72= Te6edf3ArrayListTff7b72<Te6edf3ByteArrayTff7b72>Tb4b4b4(Tff7b72if Tb4b4b4(Te6edf3countedTb4b4b4) Te6edf3expectedTb4b4b4.Te6edf3coerceAtMostTb4b4b4(Te6edf3MAX_PREALLOCTb4b4b4) Tff7b72else T79c0ff0Tb4b4b4)

Tff7b72while Tb4b4b4(Te6edf3framesTb4b4b4.Te6edf3size Tff7b72< Te6edf3expected Tff7b72&Tff7b72& Tff7b72!Te6edf3exhaustedTb4b4b4(Tb4b4b4)Tb4b4b4) Tb4b4b4{

Tff7b72val Te6edf3len Tff7b72= Te6edf3readUInt32Tb4b4b4(Ta5d6ff"Tffd700$Te6edf3labelTa5d6ff[Tffd700${Te6edf3framesTb4b4b4.Te6edf3sizeTffd700}Ta5d6ff] lengthTa5d6ff"Tb4b4b4)

Te6edf3requireTb4b4b4(Te6edf3lenTb4b4b4.Te6edf3toLongTb4b4b4(Tb4b4b4) Tff7b72<Tff7b72= Te6edf3sizeTb4b4b4) Tb4b4b4{

Ta5d6ff"Ta5d6ffMalformed replay asset: Tffd700$Te6edf3labelTa5d6ff frame Tffd700${Te6edf3framesTb4b4b4.Te6edf3sizeTffd700}Ta5d6ff declares Tffd700$Te6edf3lenTa5d6ff bytes, only Tffd700$Te6edf3sizeTa5d6ff remainTa5d6ff"

Tb4b4b4}

Te6edf3frames Tff7b72+Tff7b72= Te6edf3readByteArrayTb4b4b4(Te6edf3lenTb4b4b4.Te6edf3toLongTb4b4b4(Tb4b4b4)Tb4b4b4)

Tb4b4b4}

Te6edf3requireTb4b4b4(Tff7b72!Te6edf3counted Tff7b72|Tff7b72| Te6edf3framesTb4b4b4.Te6edf3size Tff7b72=Tff7b72= Te6edf3expectedTb4b4b4) Tb4b4b4{

Ta5d6ff"Ta5d6ffMalformed replay asset: Tffd700$Te6edf3labelTa5d6ff section truncated — read Tffd700${Te6edf3framesTb4b4b4.Te6edf3sizeTffd700}Ta5d6ff of Tffd700$Te6edf3expectedTa5d6ff framesTa5d6ff"

Tb4b4b4}

Tff7b72return Te6edf3frames

Tb4b4b4}

T8b949e/** Reads a big-endian `u32` as a non-negative [Int], rejecting truncation and the >2 GiB high-bit range. */

Tff7b72private Tff7b72fun Te6edf3BufferTb4b4b4.Td2a8ffreadUInt32Tb4b4b4(Te6edf3labelTb4b4b4: Tffa657StringTb4b4b4)Tb4b4b4: Tffa657Int Tb4b4b4{

Tff7b72if Tb4b4b4(Te6edf3size Tff7b72< Te6edf3INT_BYTESTb4b4b4) Tff7b72throw Te6edf3IllegalArgumentExceptionTb4b4b4(Ta5d6ff"Ta5d6ffMalformed replay asset: truncated Tffd700$Te6edf3labelTa5d6ff"Tb4b4b4)

Tff7b72val Te6edf3value Tff7b72=

Tff7b72try Tb4b4b4{

Te6edf3readIntTb4b4b4(Tb4b4b4)

Tb4b4b4} Tff7b72catch Tb4b4b4(Te6edf3eTb4b4b4: Te6edf3EOFExceptionTb4b4b4) Tb4b4b4{

Tff7b72throw Te6edf3IllegalArgumentExceptionTb4b4b4(Ta5d6ff"Ta5d6ffMalformed replay asset: truncated Tffd700$Te6edf3labelTa5d6ff"Tb4b4b4, Te6edf3eTb4b4b4)

Tb4b4b4}

Te6edf3requireTb4b4b4(Te6edf3value Tff7b72>Tff7b72= T79c0ff0Tb4b4b4) Tb4b4b4{ Ta5d6ff"Ta5d6ffMalformed replay asset: implausible Tffd700$Te6edf3labelTa5d6ff (Tffd700$Te6edf3valueTa5d6ff)Ta5d6ff" Tb4b4b4}

Tff7b72return Te6edf3value

Tb4b4b4}

Tff7b72companion Tff7b72object Tb4b4b4{

T8b949e/** ~10 packets/sec, matching the burningmesh-replay steady-rate rig used for ingestion perf work. */

Tff7b72const Tff7b72val Te6edf3DEFAULT_PACKET_DELAY_MS Tff7b72= T79c0ff1T79c0ff0T79c0ff0L

Tff7b72private Tff7b72const Tff7b72val Te6edf3INT_BYTES Tff7b72= T79c0ff4

T8b949e/** Cap the up-front list capacity from a (validated but still attacker-influenced) count to avoid OOM. */

Tff7b72private Tff7b72const Tff7b72val Te6edf3MAX_PREALLOC Tff7b72= T79c0ff4T79c0ff0T79c0ff9T79c0ff6

Tb4b4b4}

Tb4b4b4}